package cc.example;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    public static void main(String[] args) throws InterruptedException {


        // 发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 中间操作  给每个元素添加前缀
        MyProcessor1 myProcessor1 = new MyProcessor1();
        MyProcessor2 myProcessor2 = new MyProcessor2();
        MyProcessor3 myProcessor3 = new MyProcessor3();


        // 订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            /**
             * onXXXXX 在某某事件发生时,执行这个回调
             * @param subscription 订阅关系
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始" + subscription);
                this.subscription = subscription;

                // 从上游请求一个数据
                subscription.request(1);
            }

            /**
             * 下一个元素到达时,执行这个回调,  接收到新数据
             * @param item the item
             */
            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread() + "订阅者,接收到数据" + item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread() + "订阅者,接收到异常错误信息" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread() + "订阅者,接收到完成");
            }
        };

        // 订阅着绑定到发布者
        publisher.subscribe(myProcessor1);
        myProcessor1.subscribe(myProcessor2);
        myProcessor2.subscribe(myProcessor3);
        myProcessor3.subscribe(subscriber);



        // 发布者发布10条数据
        for (int i = 0; i < 10; i++) {
            if (i >= 9) {
//                publisher.closeExceptionally(new RuntimeException("不干了"));
            } else {
                publisher.submit("SubmissionPublisher - " + i);
            }
        }

        publisher.close();

        Thread.sleep(20000);

    }
}
